#include "app_pool.h"
#include <mqueue.h>
#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>
#include "log.h"

static pthread_t app_pool_threads[APP_POOL_THREAD_NUM];
static mqd_t mqd;
/**
 * @brief 线程函数   不停的从队列中取出任务进行处理
 */
void *thread_func(void *)
{
    app_pool_task_t pool_task;
    while(1)
    {
            int size= mq_receive(mqd, (char *)&pool_task, sizeof(app_pool_task_t),0);
            if(size>0)
            {
                int ret=  pool_task.func(pool_task.arg);
                if(ret!=0)
                {
                    log_error("task error ret=%d\n",ret);
                }
            }
    }

}


/**
 * @brief 线程池初始化
 */
void app_pool_init(void)
{
    //清除之前历史队列
    mq_unlink(POOL_QUEUE_NAME);
    // 创建队列容器
    struct mq_attr attr;
    attr.mq_maxmsg = 10;
    attr.mq_msgsize = sizeof(app_pool_task_t);
      mqd = mq_open(POOL_QUEUE_NAME, O_CREAT | O_RDWR, 0664, &attr);

    if (mqd == -1) {
        perror("mq_open error\n");
        return;
    }
    // 创建线程数组

    for (int i = 0; i < APP_POOL_THREAD_NUM; i++) {
        pthread_create(&app_pool_threads[i], NULL, thread_func , NULL);
    }
    log_info("app_pool_init ok\n");
}


/**
 * @brief 向线程池提交任务
 */
int app_pool_submitTask(int (*func)(void *), void *arg)
{
   app_pool_task_t pool_task;
   pool_task.func=func;
   pool_task.arg=arg;
   
    //向mqd中写入任务
   int ret=  mq_send(mqd, (char *)&pool_task, sizeof(app_pool_task_t),0);
   if(ret!=0)
   {
       perror("mq_send error ");
       return -1;
   }
   return 0;
}


/**
 * @brief 销毁线程池
 */
void app_pool_destroy(void)
{

    // 让线程停工 
    for (int i = 0; i < APP_POOL_THREAD_NUM; i++) {
        pthread_cancel(app_pool_threads[i]);
    }
    for (int i = 0; i < APP_POOL_THREAD_NUM; i++) {
        pthread_join(app_pool_threads[i],NULL);
    }

    //  销毁队列
    mq_close(mqd);
    mq_unlink(POOL_QUEUE_NAME);

}